import { Observable, interval, take } from "rxjs";
import { map, filter, reduce, find, findIndex } from 'rxjs/operators'

//类似于迭代器 next 发出通知  complete通知完成
const observable = new Observable((subscriber) => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    //   subscriber.complete();

    setTimeout(() => {
        subscriber.next(4);
        subscriber.complete();
    }, 2000);
})

observable.subscribe({
    next: (value) => console.log('输出: ', value),
    error: (err) => console.log(err),
    complete: () => console.log('complete')
})


// test2
// interval(1000) 创建了一个 Observable，它会每隔1秒钟发出一个递增的整数值。这个值从0开始，每次增加1。
// interval 五百毫秒执行一次 pipe 就是管道的意思 
// 管道里面也是可以去掉接口的支持处理异步数据 去处理数据 这儿展示 了 map  和 filter 跟数组的方法是一样的  最后 通过观察者  subscribe 接受回调

const subs = interval(500).pipe(map(v => ({ num: v })), filter(v => (v.num > 2))).subscribe(e => {
    console.log('e: ', e);
    if (e.num > 10) {
        // 停止
        subs.unsubscribe()
    }
})

// take控制输出次数
const subs2 = interval(500).pipe(take(10)).subscribe(e => {
    console.log('e2: ', e);
})